#pragma once
#include "../mqcommon/Log.hpp"
#include "../mqcommon/Helper.hpp"
#include "../mqcommon/Message.pb.h"

#include <iostream>
#include <string>
#include <unordered_map>
#include <memory>
#include <mutex>
#include <list>

namespace ouyangmq
{
    /*
        // 消息的基本属性: 1. ID  2. 投递模式 3. routing_key
        message BasicProperties{
            string id = 1;
            DeliveryMode delivery_mode = 2;
            string routing_key = 3;
        };

        // 消息：1. 消息的基本属性 2.
        message Message {
            message Payload {
                BasicProperties properties = 1; // 基本属性
                string body = 2;    // 主体
                string valid = 3;   // 是否有效(使用1/0表示) --> 在持久化存储删除时改变这个即可，防止多次IO, 真正删除的时候有垃圾回收解决
            };
            Payload payload = 1;    // 有效载荷
            uint32 offset = 2;      // 偏移量 --> 消息存储的起始位置
            uint32 length = 3;      // 长度  --> 读取消息内容时防止粘包问题
        };
    */

    using MessagePtr = std::shared_ptr<Message>;

    class MessageMapper
    {
    public:
#define DATAFILE_SUBFIX ".mqd"
#define TMPFILE_SUBFIX ".mqd.tmp"
        MessageMapper(const std::string &basedir, const std::string &qname)
            : _qname(qname)
        {
            // 1. 生成指定的目录下文件: basedir/_datafile.mqd  basedir/_tmpfile.mqd.tmp(不需要提前生成)
            std::string base = basedir.back() == '/' ? basedir : basedir + '/';
            _datafile = base + _qname + DATAFILE_SUBFIX;
            _tmpfile = base + _qname + TMPFILE_SUBFIX;
            // 2. void FileCreate(); 将具体的文件创建出来
            FileCreate();
        }

        // 消息文件的创建和删除
        void FileCreate()
        {
            // 1. 进行当前文件是否存在判断
            HelperModule::FileHelper data(_datafile);
            if (data.Exits() == true)
                return;
            // 2. 如果不存在再进行创建
            HelperModule::FileHelper::Createfile(_datafile);
        }

        void FileDelete()
        {
            // 1. 直接删除全部的文件: _datafile _tmpfile
            HelperModule::FileHelper::Destoryfile(_datafile);
            HelperModule::FileHelper::Destoryfile(_tmpfile);
        }

        // 消息的新增持久化/删除消息的持久化
        bool AddDurable(MessagePtr &message)
        {
            return _AddDurable(_datafile, message);
        }

        bool DeleteDurable(MessagePtr &message)
        {
            // 1. 将message中的valid改为"0" + 序列化
            message->mutable_payload()->set_valid("0");
            std::string body = message->payload().SerializeAsString();
            if (body.size() != message->length())
            {
                LogModule::LOG(LogModule::LogLevel::ERROR) << "消息长度序列化后出现异常";
                return false;
            }
            // 2. 覆盖式写入对应消息在文件中的位置
            HelperModule::FileHelper data(_datafile);
            bool ret = data.Write(body.c_str(), message->offset(), body.size());
            if (ret == false)
            {
                LogModule::LOG(LogModule::LogLevel::ERROR) << "写入消息的长度到->" << _datafile << "失败";
                return false;
            }
            return true;
        }

        // 垃圾回收/重新加载所有有效消息
        std::list<MessagePtr> gc()
        {
            bool ret;
            std::list<MessagePtr> result;
            // 1. 从_datafile读取加载消息
            ret = _load(result);
            if (ret == false)
            {
                LogModule::LOG(LogModule::LogLevel::ERROR) << "读取加载消息失败";
                return std::list<MessagePtr>();
            }
            // 2. 将有效的消息写入到_tmpfile
            HelperModule::FileHelper tmp(_tmpfile);
            for (auto &msg : result)
            {
                LogModule::LOG(LogModule::LogLevel::ERROR) << "向临时文件写入有效消息";
                ret = _AddDurable(_tmpfile, msg);
                if (ret == false)
                {
                    LogModule::LOG(LogModule::LogLevel::ERROR) << "向临时文件写入有效消息失败";
                    return result;
                }
            }
            // 3. 最后更改_tmpfile为_datafile,
            ret = HelperModule::FileHelper::Destoryfile(_datafile);
            if (ret == false)
            {
                LogModule::LOG(LogModule::LogLevel::ERROR) << "删除源文件失败!";
                return result;
            }
            ret = tmp.rename(_datafile);
            if (ret == false)
            {
                LogModule::LOG(LogModule::LogLevel::ERROR) << "更改临时文件名称失败!";
                return result;
            }
            // 4. 返回std::list<MessagePtr>
            return result;
        }

    private:
        bool _load(std::list<MessagePtr> &result)
        {
            // 1. 加载出文件中所有有效的消息
            HelperModule::FileHelper data(_datafile);
            size_t offset = 0, msg_size = 0;
            size_t fsize = data.size();
            bool ret;
            while (offset < fsize)
            {
                ret = data.Read((char *)&msg_size, offset, sizeof(size_t));
                if (ret == false)
                {
                    LogModule::LOG(LogModule::LogLevel::ERROR) << "读取消息长度失败";
                    return false;
                }
                offset += sizeof(size_t);
                // 初始化msgbody
                std::string msgbody(msg_size, '\0');
                ret = data.Read(&msgbody[0], offset, msg_size);
                if (ret == false)
                {
                    LogModule::LOG(LogModule::LogLevel::ERROR) << "读取消息主体失败";
                    return false;
                }
                offset += msg_size;
                MessagePtr msgp = std::make_shared<Message>();
                // 反序列化
                msgp->mutable_payload()->ParseFromString(msgbody);
                if (msgp->payload().valid() == "0")
                {
                    LogModule::LOG(LogModule::LogLevel::ERROR) << "跳过加载无效消息";
                    continue;
                }
                result.push_back(msgp);
            }
            return true;
        }
        bool _AddDurable(const std::string filename, MessagePtr &message)
        {
            // 1. 取出文件的末尾
            HelperModule::FileHelper data(_datafile);
            size_t fsize = data.size();
            // 2. 将消息序列化
            std::string body = message->payload().SerializeAsString();
            size_t bsize = body.size();
            // 3. 写入文件之中
            // 长度/主体|长度/主体
            // [1] 先将消息的长度写入(4字节)
            // [2] 再写入消息的主体
            bool ret = data.Write((char *)&bsize, fsize, sizeof(size_t));
            if (ret == false)
            {
                LogModule::LOG(LogModule::LogLevel::ERROR) << "写入消息的长度到->" << filename << "失败";
                return false;
            }

            ret = data.Write(body.c_str(), fsize + sizeof(size_t), bsize);
            if (ret == false)
            {
                LogModule::LOG(LogModule::LogLevel::ERROR) << "写入消息的主体到->" << filename << "失败";
                return false;
            }
            // 4. 更新消息的内容
            message->set_offset(fsize + sizeof(size_t));
            message->set_length(bsize);
            return true;
        }
        std::string _qname;
        std::string _datafile;
        std::string _tmpfile;
    };

    class QueueMessage
    {
    public:
        QueueMessage(const std::string &basedir, const std::string &qname)
            : _mapper(basedir, qname),
              _qname(qname),
              _valid_count(0),
              _total_count(0)
        {
        }

        // 新增消息
        bool AddMessage(const BasicProperties *bp, const std::string &body, bool queue_is_durable)
        {
            // 1. 构造一个消息对象MessagePtr
            //   [1] 用户给了bp, bp != nullptr
            //   [2] 用户没给bp, bp == nullptr
            MessagePtr msg = std::make_shared<Message>();
            msg->mutable_payload()->set_body(body);
            DeliveryMode mode = queue_is_durable == true ? DeliveryMode::DURABLE : DeliveryMode::UNDURABLE;
            if (bp != nullptr)
            {
                msg->mutable_payload()->mutable_properties()->set_id(bp->id());
                msg->mutable_payload()->mutable_properties()->set_delivery_mode(mode);
                msg->mutable_payload()->mutable_properties()->set_routing_key(bp->routing_key());
            }
            else
            {
                msg->mutable_payload()->mutable_properties()->set_id(HelperModule::UUIDHelper::GetUUID());
                msg->mutable_payload()->mutable_properties()->set_delivery_mode(mode);
                msg->mutable_payload()->mutable_properties()->set_routing_key("");
            }
            // 加锁: 访问临界资源
            std::unique_lock<std::mutex> lock(_mutex);
            // 2. 查看该消息基本属性的投递模式是否为持久的
            if (mode == DeliveryMode::DURABLE)
            {
                //   [1] 如果是持久的, 设置valid("1"), 进行持久化存储, 添加到_waitpush_msgs和_durable_msgs, ++_valid_count和_total_count
                msg->mutable_payload()->set_valid("1");
                bool ret = _mapper.AddDurable(msg);
                if(ret == false)
                {
                    LogModule::LOG(LogModule::LogLevel::ERROR) << "持久化存储失败: " << body.c_str();
                    return false;
                }
                _durable_msgs.emplace(msg->payload().properties().id(), msg);
                _valid_count += 1; 
                _total_count += 1;
            }
            //   [2] 如果不是持久的, 添加到_waitpush_msgs
            _waitpush_msgs.push_back(msg);
            return true;
        }

        // 获取队首消息
        MessagePtr Front()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            // 1. 判断_waitpush_msgs是否存在消息
            if(_waitpush_msgs.size() == 0)
            {
                // 如果待推送消息链表没有消息, 返回一个空消息
                return MessagePtr();
            }
            // 2. 取出队首消息, 删除队首消息
            MessagePtr msg = _waitpush_msgs.front();
            _waitpush_msgs.pop_front();
            // 3. 向待确认消息的hashmap添加一份
            _waitack_msgs.emplace(msg->payload().properties().id(), msg);
            return msg;
        }

        // 确认消息(删除消息) + 垃圾回收条件判断, 满足则执行内部的垃圾gc()回收函数进行垃圾回收
        bool RomveMessage(const std::string &id)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            // 1. 现在对应的待确认消息的hashmap查找要删除消息的id
            auto it = _waitack_msgs.find(id);
            if(it == _waitack_msgs.end())
            {
                // 如果没有找到说明这条消息已经被删除了
                LogModule::LOG(LogModule::LogLevel::ERROR) << "没有找到你要删除的消息id: " << id;
                return true;
            }
            // 2. 根据消息持久化判断是否要删除持久化存储, 同时检测垃圾回收机制条件判断
            if(it->second->payload().properties().delivery_mode() == DeliveryMode::DURABLE)
            {
                _mapper.DeleteDurable(it->second);
                _durable_msgs.erase(id);
                _valid_count -= 1;//持久化文件中有效消息数量 -1
                gc();  //内部判断是否需要垃圾回收，需要的话则回收一下
            }
            // 3. 内存中删除消息
            _waitack_msgs.erase(id);
            return true;
        }


        // 删除队列所有消息
        void clear()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            _mapper.FileDelete();
            _waitpush_msgs.clear();
            _durable_msgs.clear();
            _waitack_msgs.clear();
            _valid_count = _total_count = 0;
        }

        // 获取待推送消息的数量
        size_t Get_waitpush_count()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            return _waitpush_msgs.size();
        }

        // 获取待确认消息数量
        size_t Get_waitack_count()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            return _waitack_msgs.size();
        }

        // 获取持久化消息数量
        size_t Get_duarbale_count()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            return _durable_msgs.size();
        }

        // 获取总体消息数量
        size_t Get_total_count()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            return _total_count;
        }

        // 获取有效消息数量
        size_t Get_valid_count()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            return _valid_count;
        }

    private:
        void gc()
        {
            // 1. 如果消息数量 > 2000 && 有效消息数的占比低于50% 进行垃圾回收
            if(!(_total_count > 2000 && (_valid_count * 10) / _total_count < 5)) return;
            // 2. 获取有效的消息链表
            std::list<MessagePtr> msgs = _mapper.gc();
            // 3. 遍历有效消息链表, 重新更新实际存储位置
            for(auto &msg : msgs)
            {
                auto it = _durable_msgs.find(msg->payload().properties().id());
                if(it == _durable_msgs.end())
                {
                    LogModule::LOG(LogModule::LogLevel::ERROR) << "垃圾回收发现一条持久化消息没有在内存中进行管理!";
                    _waitpush_msgs.push_back(msg);
                    _durable_msgs.emplace(msg->payload().properties().id(), msg);
                    continue;
                }

                // 重新更新实际存储位置
                it->second->set_offset(msg->offset());
                it->second->set_length(msg->length());
            }
            // 4. 更新当前有效消息数量和总消息数量
            _valid_count = _total_count = msgs.size();
        }

    private:
        // 互斥锁
        std::mutex _mutex;
        // 队列名称
        std::string _qname;
        // 持久化管理的消息句柄
        MessageMapper _mapper;
        // 待推送消息链表
        std::list<MessagePtr> _waitpush_msgs;
        // 持久化消息的hashmap
        std::unordered_map<std::string, MessagePtr> _durable_msgs;
        // 待确认消息的hashmap
        std::unordered_map<std::string, MessagePtr> _waitack_msgs;
        // 有效消息数量
        size_t _valid_count;
        // 总体消息数量
        size_t _total_count;
    };

}